package com.xxl.job.admin.core.thread;

import com.alibaba.fastjson2.JSON;
import com.google.gson.Gson;
import com.xxl.job.admin.core.complete.XxlJobCompleter;
import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
import com.xxl.job.admin.core.model.XxlJobLog;
import com.xxl.job.admin.core.util.I18nUtil;
import com.xxl.job.admin.core.util.SqlSessionUtil;
import com.xxl.job.core.biz.model.HandleCallbackParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.util.DateUtil;
import org.apache.ibatis.session.SqlSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * job lose-monitor instance
 *
 * @author xuxueli 2015-9-1 18:05:56
 */
public class JobCompleteHelper {
    private static Logger logger = LoggerFactory.getLogger(JobCompleteHelper.class);

    private static JobCompleteHelper instance = new JobCompleteHelper();

    public static JobCompleteHelper getInstance() {
        return instance;
    }

    // ---------------------- monitor ----------------------

    private ThreadPoolExecutor callbackThreadPool = null;
    private Thread monitorThread;
    private volatile boolean toStop = false;

    public void start() {

        // for callback
        callbackThreadPool = new ThreadPoolExecutor(
                10,
                20,
                1,
                TimeUnit.HOURS,
                new LinkedBlockingQueue<>(100000),
                r -> new Thread(r, "xxl-job, admin JobLosedMonitorHelper-callbackThreadPool-" + r.hashCode()),
                (r, executor) -> {
                    r.run();
                    logger.warn(">>>>>>>>>>> xxl-job, callback too fast, match threadpool rejected handler(run now).");
                });

        callbackThreadPool.prestartCoreThread();


        // for monitor
        monitorThread = new Thread(() -> {

            // wait for JobTriggerPoolHelper-init
            try {
                TimeUnit.MILLISECONDS.sleep(50);
            } catch (InterruptedException e) {
                if (!toStop) {
                    logger.error(e.getMessage(), e);
                }
            }

            // monitor
            while (!toStop) {
                try {
                    // 任务结果丢失处理：调度记录停留在 "运行中" 状态超过10min，且对应执行器心跳注册失败不在线，则将本地调度主动标记失败；
                    Date losedTime = DateUtil.addMinutes(new Date(), -10);
                    List<Long> losedJobIds = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().findLostJobIds(losedTime);

                    //todo 优化 如果设置了任务执行超时时间大于10分钟则过滤此类log
                    if (losedJobIds != null && losedJobIds.size() > 0) {
                        for (Long logId : losedJobIds) {

                            XxlJobLog jobLog = new XxlJobLog();
                            jobLog.setId(logId);

                            jobLog.setHandleTime(new Date());
                            jobLog.setHandleCode(ReturnT.FAIL_CODE);
                            jobLog.setHandleMsg(I18nUtil.getString("joblog_lost_fail"));

                            XxlJobCompleter.updateHandleInfoAndFinish(jobLog);
                        }

                    }
                } catch (Exception e) {
                    if (!toStop) {
                        logger.error(">>>>>>>>>>> xxl-job, job fail monitor thread error:{}", e);
                    }
                }

                try {
                    TimeUnit.SECONDS.sleep(60);
                } catch (Exception e) {
                    if (!toStop) {
                        logger.error(e.getMessage(), e);
                    }
                }

            }

            logger.info(">>>>>>>>>>> xxl-job, JobLosedMonitorHelper stop");

        });
        monitorThread.setDaemon(true);
        monitorThread.setName("xxl-job, admin JobLosedMonitorHelper");
        monitorThread.start();
    }

    public void toStop() {
        toStop = true;

        // stop registryOrRemoveThreadPool
        callbackThreadPool.shutdownNow();

        // stop monitorThread (interrupt and wait)
        monitorThread.interrupt();
        try {
            monitorThread.join();
        } catch (InterruptedException e) {
            logger.error(e.getMessage(), e);
        }
    }


    // ---------------------- helper ----------------------

    /**
     * 处理回调
     * @param callbackParamList
     * @return
     */
    public ReturnT<String> callback(List<HandleCallbackParam> callbackParamList) {

        callbackThreadPool.execute(() -> {
            logger.info("callBack size = {}",callbackParamList.size());
            List<XxlJobLog> xxlJobLogList = new ArrayList<>();

            for (HandleCallbackParam handleCallbackParam : callbackParamList) {
                xxlJobLogList.add(getCallbackLog(handleCallbackParam));
            }
            callbackList(xxlJobLogList);
        });

        return ReturnT.SUCCESS;
    }

    @Deprecated
    private ReturnT<String> callback(HandleCallbackParam handleCallbackParam) {

        long startTime = System.currentTimeMillis(), endTime;
        XxlJobLog log;
        // valid log item,这里可以通过缓存log id来判断是否存在重复回调，减少一次数据库IO
        if(handleCallbackParam.getLogId() > 0L){
            log = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().load(handleCallbackParam.getLogId());
        }else {
            log = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().loadByUUID(handleCallbackParam.getLogUUID());
        }
        if (log == null) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "log item not found.");
        }
        if (log.getHandleCode() > 0) {
            // avoid repeat callback, trigger child job etc
            return new ReturnT<>(ReturnT.FAIL_CODE, "log repeate callback.");
        }

        // handle msg
        StringBuffer handleMsg = new StringBuffer();
        if (log.getHandleMsg() != null) {
            handleMsg.append(log.getHandleMsg()).append("<br>");
        }
        if (handleCallbackParam.getHandleMsg() != null) {
            handleMsg.append(handleCallbackParam.getHandleMsg()).append("<br>");
        }

        log.setHandleCode(handleCallbackParam.getHandleCode());
        log.setHandleMsg(handleMsg.toString());
        if (log.getHandleMsg().length() > 15000) {
            log.setHandleMsg( log.getHandleMsg().substring(0, 15000) );
        }

        XxlJobCompleter.updateHandleInfoAndFinish(log);
        endTime = System.currentTimeMillis();
        if(endTime - startTime > 50){
            logger.info("dbTime = {}",(endTime - startTime));
        }

        return ReturnT.SUCCESS;
    }


    /**
     * 批量修改回调日志
     * @param xxlJobLogList
     * @return
     */
    private ReturnT<String> callbackList(List<XxlJobLog> xxlJobLogList) {

        List<XxlJobLog> xxlJobLogList1 = xxlJobLogList.stream().filter(xxlJobLog -> xxlJobLog.getUuid() != null && xxlJobLog.getUuid().length() > 0).collect(Collectors.toList());
        List<XxlJobLog> xxlJobLogList2 = xxlJobLogList.stream().filter(xxlJobLog -> xxlJobLog.getId() > 0L).collect(Collectors.toList());

        SqlSession session = SqlSessionUtil.createSqlBatchSession();

        String updateMethodStatementUUId = "com.xxl.job.admin.dao.XxlJobLogDao.updateHandleInfoBatchByUUID";
        String updateMethodStatementId = "com.xxl.job.admin.dao.XxlJobLogDao.updateHandleInfoBatchById";

        if(!xxlJobLogList1.isEmpty()){
            for (XxlJobLog xxlJobLog : xxlJobLogList1){
                session.update(updateMethodStatementUUId, xxlJobLog);
            }
        }
        if(!xxlJobLogList2.isEmpty()){
            for (XxlJobLog xxlJobLog : xxlJobLogList2){
                session.update(updateMethodStatementId, xxlJobLog);
            }
        }
        SqlSessionUtil.flush(session);
        return ReturnT.SUCCESS;
    }

    /**
     * 构建修改xxlJoblog对象
     * @param handleCallbackParam
     * @return
     */
    private XxlJobLog getCallbackLog(HandleCallbackParam handleCallbackParam){
        XxlJobLog log = new XxlJobLog();
        log.setJobId(handleCallbackParam.getLogId());
        log.setUuid(handleCallbackParam.getLogUUID());
        log.setHandleTime(new Date(handleCallbackParam.getStartExeTime()));
        log.setHandleCode(handleCallbackParam.getHandleCode());

        // handle msg
        StringBuffer handleMsg = new StringBuffer();
        if (log.getHandleMsg() != null) {
            handleMsg.append(log.getHandleMsg()).append("<br>");
        }
        if (handleCallbackParam.getHandleMsg() != null) {
            handleMsg.append(handleCallbackParam.getHandleMsg()).append("<br>");
            log.setHandleMsg(handleMsg.toString());
            if (log.getHandleMsg().length() > 15000) {
                log.setHandleMsg( log.getHandleMsg().substring(0, 15000) );
            }
        }else {
            log.setHandleMsg("业务执行耗时"+handleCallbackParam.getCost()+"ms");
        }
        return log;
    }


}
